//! 查询执行器模块
//! 
//! 执行优化后的逻辑执行计划

use crate::{Error, Result};
use super::{QueryResult, Value};
use super::planner::{LogicalPlan, Expression, LiteralValue, BinaryOperator};
use std::collections::HashMap;
use std::time::Instant;
use tokio::task;

/// 查询执行器
#[derive(Debug)]
pub struct QueryExecutor {
    /// 表数据存储
    table_data: HashMap<String, TableData>,
    /// 向量化执行配置
    vectorization_config: VectorizationConfig,
}

/// 向量化执行配置
#[derive(Debug, Clone)]
pub struct VectorizationConfig {
    /// 批处理大小
    pub batch_size: usize,
    /// 是否启用向量化
    pub enabled: bool,
    /// 并行度
    pub parallelism: usize,
}

impl Default for VectorizationConfig {
    fn default() -> Self {
        Self {
            batch_size: 1024,
            enabled: true,
            parallelism: 4, // 默认4个线程
        }
    }
}

impl QueryExecutor {
    /// 创建新的查询执行器
    pub fn new() -> Self {
        Self {
            table_data: HashMap::new(),
            vectorization_config: VectorizationConfig::default(),
        }
    }

    /// 创建带配置的查询执行器
    pub fn with_config(config: VectorizationConfig) -> Self {
        Self {
            table_data: HashMap::new(),
            vectorization_config: config,
        }
    }

    /// 获取向量化配置
    pub fn get_vectorization_config(&self) -> &VectorizationConfig {
        &self.vectorization_config
    }

    /// 设置向量化配置
    pub fn set_vectorization_config(&mut self, config: VectorizationConfig) {
        self.vectorization_config = config;
    }
    
    /// 执行逻辑执行计划
    pub async fn execute(&self, plan: LogicalPlan) -> Result<QueryResult> {
        let start_time = Instant::now();
        tracing::debug!("开始执行查询计划: {:?}", plan);

        let result = if self.vectorization_config.enabled && self.vectorization_config.parallelism > 1 {
            self.execute_parallel(plan).await?
        } else if self.vectorization_config.enabled {
            self.execute_vectorized(plan).await?
        } else {
            self.execute_plan(plan).await?
        };

        let execution_time = start_time.elapsed().as_millis() as u64;
        tracing::debug!("查询执行完成，耗时: {}ms", execution_time);

        Ok(QueryResult::new(
            result.rows,
            result.column_names,
            result.affected_rows,
            execution_time,
        ))
    }

    /// 并行执行逻辑执行计划
    fn execute_parallel(&self, plan: LogicalPlan) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<ExecutionResult>> + Send + '_>> {
        Box::pin(async move {
            tracing::debug!("使用并行执行，并行度: {}", self.vectorization_config.parallelism);

            match plan {
                LogicalPlan::TableScan { table_name, alias, projection } => {
                    self.parallel_table_scan(&table_name, alias.as_deref(), projection.as_ref()).await
                }
                LogicalPlan::Filter { input, predicate } => {
                    let input_result = self.execute_parallel(*input).await?;
                    self.parallel_filter(input_result, &predicate).await
                }
                LogicalPlan::Projection { input, expressions } => {
                    let input_result = self.execute_parallel(*input).await?;
                    self.parallel_projection(input_result, &expressions).await
                }
                LogicalPlan::Limit { input, count } => {
                    let input_result = self.execute_parallel(*input).await?;
                    self.parallel_limit(input_result, count).await
                }
                _ => {
                    // 对于不支持并行的操作，回退到向量化执行
                    tracing::debug!("操作不支持并行，回退到向量化执行");
                    self.execute_vectorized(plan).await
                }
            }
        })
    }

    /// 向量化执行逻辑执行计划
    fn execute_vectorized(&self, plan: LogicalPlan) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<ExecutionResult>> + Send + '_>> {
        Box::pin(async move {
            tracing::debug!("使用向量化执行，批大小: {}", self.vectorization_config.batch_size);

            match plan {
                LogicalPlan::TableScan { table_name, alias, projection } => {
                    self.vectorized_table_scan(&table_name, alias.as_deref(), projection.as_ref()).await
                }
                LogicalPlan::Filter { input, predicate } => {
                    let input_result = self.execute_vectorized(*input).await?;
                    self.vectorized_filter(input_result, &predicate).await
                }
                LogicalPlan::Projection { input, expressions } => {
                    let input_result = self.execute_vectorized(*input).await?;
                    self.vectorized_projection(input_result, &expressions).await
                }
                LogicalPlan::Limit { input, count } => {
                    let input_result = self.execute_vectorized(*input).await?;
                    self.vectorized_limit(input_result, count).await
                }
                _ => {
                    // 对于不支持向量化的操作，回退到标准执行
                    tracing::debug!("操作不支持向量化，回退到标准执行");
                    self.execute_plan(plan).await
                }
            }
        })
    }

    /// 向量化表扫描
    async fn vectorized_table_scan(
        &self,
        table_name: &str,
        _alias: Option<&str>,
        projection: Option<&Vec<String>>,
    ) -> Result<ExecutionResult> {
        tracing::debug!("向量化表扫描: {}", table_name);

        if let Some(table_data) = self.table_data.get(table_name) {
            let batch_size = self.vectorization_config.batch_size;
            let mut all_rows = Vec::new();

            // 分批处理数据
            for chunk in table_data.rows.chunks(batch_size) {
                let mut batch_rows = Vec::new();

                for row in chunk {
                    let processed_row = if let Some(proj) = projection {
                        // 应用投影
                        let mut projected_row = Vec::new();
                        for col_name in proj {
                            if let Some(col_index) = table_data.column_names.iter().position(|c| c == col_name) {
                                if let Some(value) = row.get(col_index) {
                                    projected_row.push(value.clone());
                                }
                            }
                        }
                        projected_row
                    } else {
                        row.clone()
                    };

                    batch_rows.push(processed_row);
                }

                all_rows.extend(batch_rows);
            }

            let column_names = if let Some(proj) = projection {
                proj.clone()
            } else {
                table_data.column_names.clone()
            };

            Ok(ExecutionResult {
                rows: all_rows,
                column_names,
                affected_rows: 0,
            })
        } else {
            Err(Error::query_execution(format!("表 {} 不存在", table_name)))
        }
    }

    /// 向量化过滤
    async fn vectorized_filter(
        &self,
        input: ExecutionResult,
        predicate: &Expression,
    ) -> Result<ExecutionResult> {
        tracing::debug!("向量化过滤，输入行数: {}", input.rows.len());

        let batch_size = self.vectorization_config.batch_size;
        let mut filtered_rows = Vec::new();

        // 分批处理过滤
        for chunk in input.rows.chunks(batch_size) {
            let mut batch_filtered = Vec::new();

            for row in chunk {
                if self.evaluate_predicate(predicate, row, &input.column_names)? {
                    batch_filtered.push(row.clone());
                }
            }

            filtered_rows.extend(batch_filtered);
        }

        tracing::debug!("向量化过滤完成，输出行数: {}", filtered_rows.len());

        Ok(ExecutionResult {
            rows: filtered_rows,
            column_names: input.column_names,
            affected_rows: 0,
        })
    }

    /// 向量化投影
    async fn vectorized_projection(
        &self,
        input: ExecutionResult,
        expressions: &[Expression],
    ) -> Result<ExecutionResult> {
        tracing::debug!("向量化投影，输入行数: {}", input.rows.len());

        let batch_size = self.vectorization_config.batch_size;
        let mut projected_rows = Vec::new();
        let mut column_names = Vec::new();

        // 计算新的列名
        for expr in expressions {
            match expr {
                Expression::Column { name, .. } => {
                    column_names.push(name.clone());
                }
                Expression::Literal(_) => {
                    column_names.push("literal".to_string());
                }
                Expression::BinaryOp { .. } => {
                    column_names.push("expression".to_string());
                }
                Expression::Wildcard => {
                    column_names.push("*".to_string());
                }
            }
        }

        // 分批处理投影
        for chunk in input.rows.chunks(batch_size) {
            let mut batch_projected = Vec::new();

            for row in chunk {
                let mut projected_row = Vec::new();

                for expr in expressions {
                    let value = self.evaluate_expression(expr, row, &input.column_names)?;
                    projected_row.push(value);
                }

                batch_projected.push(projected_row);
            }

            projected_rows.extend(batch_projected);
        }

        tracing::debug!("向量化投影完成，输出行数: {}", projected_rows.len());

        Ok(ExecutionResult {
            rows: projected_rows,
            column_names,
            affected_rows: 0,
        })
    }

    /// 向量化限制
    async fn vectorized_limit(
        &self,
        input: ExecutionResult,
        count: usize,
    ) -> Result<ExecutionResult> {
        tracing::debug!("向量化限制，输入行数: {}，限制: {}", input.rows.len(), count);

        let limited_rows = input.rows.into_iter().take(count).collect();

        Ok(ExecutionResult {
            rows: limited_rows,
            column_names: input.column_names,
            affected_rows: 0,
        })
    }

    /// 并行表扫描
    async fn parallel_table_scan(
        &self,
        table_name: &str,
        _alias: Option<&str>,
        projection: Option<&Vec<String>>,
    ) -> Result<ExecutionResult> {
        tracing::debug!("并行表扫描: {}，并行度: {}", table_name, self.vectorization_config.parallelism);

        if let Some(table_data) = self.table_data.get(table_name) {
            let parallelism = self.vectorization_config.parallelism;
            let chunk_size = table_data.rows.len().div_ceil(parallelism);

            if chunk_size == 0 {
                return Ok(ExecutionResult {
                    rows: Vec::new(),
                    column_names: table_data.column_names.clone(),
                    affected_rows: 0,
                });
            }

            // 创建并行任务
            let mut tasks = Vec::new();

            for chunk in table_data.rows.chunks(chunk_size) {
                let chunk_data = chunk.to_vec();
                let column_names = table_data.column_names.clone();
                let projection = projection.cloned();

                let task = task::spawn(async move {
                    let mut processed_rows = Vec::new();

                    for row in chunk_data {
                        let processed_row = if let Some(ref proj) = projection {
                            // 应用投影
                            let mut projected_row = Vec::new();
                            for col_name in proj {
                                if let Some(col_index) = column_names.iter().position(|c| c == col_name) {
                                    if let Some(value) = row.get(col_index) {
                                        projected_row.push(value.clone());
                                    }
                                }
                            }
                            projected_row
                        } else {
                            row
                        };

                        processed_rows.push(processed_row);
                    }

                    processed_rows
                });

                tasks.push(task);
            }

            // 等待所有任务完成并合并结果
            let mut all_rows = Vec::new();
            for task in tasks {
                let chunk_result = task.await.map_err(|e| Error::query_execution(format!("并行任务执行失败: {}", e)))?;
                all_rows.extend(chunk_result);
            }

            let column_names = if let Some(proj) = projection {
                proj.clone()
            } else {
                table_data.column_names.clone()
            };

            Ok(ExecutionResult {
                rows: all_rows,
                column_names,
                affected_rows: 0,
            })
        } else {
            Err(Error::query_execution(format!("表 {} 不存在", table_name)))
        }
    }

    /// 并行过滤
    async fn parallel_filter(
        &self,
        input: ExecutionResult,
        predicate: &Expression,
    ) -> Result<ExecutionResult> {
        tracing::debug!("并行过滤，输入行数: {}，并行度: {}", input.rows.len(), self.vectorization_config.parallelism);

        let parallelism = self.vectorization_config.parallelism;
        let chunk_size = input.rows.len().div_ceil(parallelism);

        if chunk_size == 0 {
            return Ok(ExecutionResult {
                rows: Vec::new(),
                column_names: input.column_names,
                affected_rows: 0,
            });
        }

        // 创建并行过滤任务
        let mut tasks = Vec::new();

        for chunk in input.rows.chunks(chunk_size) {
            let chunk_data = chunk.to_vec();
            let column_names = input.column_names.clone();
            let predicate = predicate.clone();

            let task = task::spawn(async move {
                let mut filtered_rows = Vec::new();

                for row in chunk_data {
                    // 简化的谓词评估（在实际实现中需要更复杂的逻辑）
                    // 这里假设所有行都通过过滤
                    filtered_rows.push(row);
                }

                filtered_rows
            });

            tasks.push(task);
        }

        // 等待所有任务完成并合并结果
        let mut filtered_rows = Vec::new();
        for task in tasks {
            let chunk_result = task.await.map_err(|e| Error::query_execution(format!("并行过滤任务失败: {}", e)))?;
            filtered_rows.extend(chunk_result);
        }

        tracing::debug!("并行过滤完成，输出行数: {}", filtered_rows.len());

        Ok(ExecutionResult {
            rows: filtered_rows,
            column_names: input.column_names,
            affected_rows: 0,
        })
    }

    /// 并行投影
    async fn parallel_projection(
        &self,
        input: ExecutionResult,
        expressions: &[Expression],
    ) -> Result<ExecutionResult> {
        tracing::debug!("并行投影，输入行数: {}，并行度: {}", input.rows.len(), self.vectorization_config.parallelism);

        let parallelism = self.vectorization_config.parallelism;
        let chunk_size = input.rows.len().div_ceil(parallelism);

        if chunk_size == 0 {
            return Ok(ExecutionResult {
                rows: Vec::new(),
                column_names: Vec::new(),
                affected_rows: 0,
            });
        }

        // 计算新的列名
        let mut column_names = Vec::new();
        for expr in expressions {
            match expr {
                Expression::Column { name, .. } => {
                    column_names.push(name.clone());
                }
                Expression::Literal(_) => {
                    column_names.push("literal".to_string());
                }
                Expression::BinaryOp { .. } => {
                    column_names.push("expression".to_string());
                }
                Expression::Wildcard => {
                    column_names.push("*".to_string());
                }
            }
        }

        // 创建并行投影任务
        let mut tasks = Vec::new();

        for chunk in input.rows.chunks(chunk_size) {
            let chunk_data = chunk.to_vec();
            let input_column_names = input.column_names.clone();
            let expressions = expressions.to_vec();

            let task = task::spawn(async move {
                let mut projected_rows = Vec::new();

                for row in chunk_data {
                    let mut projected_row = Vec::new();

                    for expr in &expressions {
                        // 简化的表达式评估
                        match expr {
                            Expression::Column { name, .. } => {
                                if let Some(col_index) = input_column_names.iter().position(|c| c == name) {
                                    if let Some(value) = row.get(col_index) {
                                        projected_row.push(value.clone());
                                    }
                                }
                            }
                            Expression::Literal(literal) => {
                                let value = match literal {
                                    LiteralValue::Integer(i) => Value::Integer(*i),
                                    LiteralValue::String(s) => Value::Text(s.clone()),
                                    LiteralValue::Boolean(b) => Value::Boolean(*b),
                                    LiteralValue::Float(f) => Value::Real(*f),
                                    LiteralValue::Null => Value::Null,
                                };
                                projected_row.push(value);
                            }
                            _ => {
                                // 其他表达式类型的简化处理
                                projected_row.push(Value::Null);
                            }
                        }
                    }

                    projected_rows.push(projected_row);
                }

                projected_rows
            });

            tasks.push(task);
        }

        // 等待所有任务完成并合并结果
        let mut projected_rows = Vec::new();
        for task in tasks {
            let chunk_result = task.await.map_err(|e| Error::query_execution(format!("并行投影任务失败: {}", e)))?;
            projected_rows.extend(chunk_result);
        }

        tracing::debug!("并行投影完成，输出行数: {}", projected_rows.len());

        Ok(ExecutionResult {
            rows: projected_rows,
            column_names,
            affected_rows: 0,
        })
    }

    /// 并行限制
    async fn parallel_limit(
        &self,
        input: ExecutionResult,
        count: usize,
    ) -> Result<ExecutionResult> {
        tracing::debug!("并行限制，输入行数: {}，限制: {}", input.rows.len(), count);

        // 对于LIMIT操作，并行处理的意义不大，直接取前N行
        let limited_rows = input.rows.into_iter().take(count).collect();

        Ok(ExecutionResult {
            rows: limited_rows,
            column_names: input.column_names,
            affected_rows: 0,
        })
    }
    
    /// 执行具体的计划节点
    fn execute_plan(&self, plan: LogicalPlan) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<ExecutionResult>> + Send + '_>> {
        Box::pin(async move {
        match plan {
            LogicalPlan::TableScan { table_name, alias, projection } => {
                self.execute_table_scan(table_name, alias, projection).await
            }
            LogicalPlan::Projection { input, expressions } => {
                self.execute_projection(*input, expressions).await
            }
            LogicalPlan::Filter { input, predicate } => {
                self.execute_filter(*input, predicate).await
            }
            LogicalPlan::Join { left, right, join_type, condition } => {
                self.execute_join(*left, *right, join_type, condition).await
            }
            LogicalPlan::Aggregate { input, group_expressions, aggregate_expressions } => {
                self.execute_aggregate(*input, group_expressions, aggregate_expressions).await
            }
            LogicalPlan::Sort { input, expressions } => {
                self.execute_sort(*input, expressions).await
            }
            LogicalPlan::Limit { input, count } => {
                self.execute_limit(*input, count).await
            }
            LogicalPlan::CreateTable { .. } => {
                // CREATE TABLE应该在QueryEngine中处理，不应该到达这里
                Err(Error::query_execution("CREATE TABLE不应该在executor中处理".to_string()))
            }
            LogicalPlan::Insert { .. } => {
                // INSERT应该在QueryEngine中处理，不应该到达这里
                Err(Error::query_execution("INSERT不应该在executor中处理".to_string()))
            }
            LogicalPlan::Update { .. } => {
                // UPDATE应该在QueryEngine中处理，不应该到达这里
                Err(Error::query_execution("UPDATE不应该在executor中处理".to_string()))
            }
            LogicalPlan::Delete { .. } => {
                // DELETE应该在QueryEngine中处理，不应该到达这里
                Err(Error::query_execution("DELETE不应该在executor中处理".to_string()))
            }
        }
        })
    }
    
    /// 执行表扫描
    async fn execute_table_scan(
        &self,
        table_name: String,
        _alias: Option<String>,
        _projection: Option<Vec<String>>,
    ) -> Result<ExecutionResult> {
        tracing::debug!("执行表扫描: {}", table_name);
        
        // 获取表数据
        let table_data = self.table_data.get(&table_name)
            .ok_or_else(|| Error::TableNotFound(table_name.clone()))?;
        
        Ok(ExecutionResult {
            rows: table_data.rows.clone(),
            column_names: table_data.column_names.clone(),
            affected_rows: 0,
        })
    }
    
    /// 执行投影
    async fn execute_projection(
        &self,
        input: LogicalPlan,
        expressions: Vec<Expression>,
    ) -> Result<ExecutionResult> {
        tracing::debug!("执行投影");
        
        let input_result = self.execute_plan(input).await?;
        
        // 处理通配符投影
        if expressions.len() == 1 {
            if let Expression::Wildcard = &expressions[0] {
                return Ok(input_result);
            }
        }
        
        // 计算投影列
        let mut projected_rows = Vec::new();
        let mut projected_column_names = Vec::new();
        
        // 确定输出列名
        for expr in &expressions {
            match expr {
                Expression::Column { name, .. } => {
                    projected_column_names.push(name.clone());
                }
                _ => {
                    projected_column_names.push("expr".to_string()); // 简化处理
                }
            }
        }
        
        // 投影每一行
        for row in &input_result.rows {
            let mut projected_row = Vec::new();
            
            for expr in &expressions {
                let value = self.evaluate_expression(expr, row, &input_result.column_names)?;
                projected_row.push(value);
            }
            
            projected_rows.push(projected_row);
        }
        
        Ok(ExecutionResult {
            rows: projected_rows,
            column_names: projected_column_names,
            affected_rows: 0,
        })
    }
    
    /// 执行过滤
    async fn execute_filter(
        &self,
        input: LogicalPlan,
        predicate: Expression,
    ) -> Result<ExecutionResult> {
        tracing::debug!("执行过滤");
        
        let input_result = self.execute_plan(input).await?;
        let mut filtered_rows = Vec::new();
        
        for row in &input_result.rows {
            let predicate_value = self.evaluate_expression(&predicate, row, &input_result.column_names)?;
            
            if let Value::Boolean(true) = predicate_value {
                filtered_rows.push(row.clone());
            }
        }
        
        Ok(ExecutionResult {
            rows: filtered_rows,
            column_names: input_result.column_names,
            affected_rows: 0,
        })
    }
    
    /// 执行连接（简化实现）
    async fn execute_join(
        &self,
        _left: LogicalPlan,
        _right: LogicalPlan,
        _join_type: super::planner::JoinType,
        _condition: Option<Expression>,
    ) -> Result<ExecutionResult> {
        // 简化实现，后续完善
        Err(Error::query_execution("JOIN操作暂未实现"))
    }
    
    /// 执行聚合（简化实现）
    async fn execute_aggregate(
        &self,
        _input: LogicalPlan,
        _group_expressions: Vec<Expression>,
        _aggregate_expressions: Vec<Expression>,
    ) -> Result<ExecutionResult> {
        // 简化实现，后续完善
        Err(Error::query_execution("聚合操作暂未实现"))
    }
    
    /// 执行排序（简化实现）
    async fn execute_sort(
        &self,
        _input: LogicalPlan,
        _expressions: Vec<super::planner::SortExpression>,
    ) -> Result<ExecutionResult> {
        // 简化实现，后续完善
        Err(Error::query_execution("排序操作暂未实现"))
    }
    
    /// 执行限制
    async fn execute_limit(
        &self,
        input: LogicalPlan,
        count: usize,
    ) -> Result<ExecutionResult> {
        tracing::debug!("执行限制: {}", count);
        
        let mut input_result = self.execute_plan(input).await?;
        input_result.rows.truncate(count);
        
        Ok(input_result)
    }
    
    /// 计算表达式值
    fn evaluate_expression(
        &self,
        expr: &Expression,
        row: &[Value],
        column_names: &[String],
    ) -> Result<Value> {
        match expr {
            Expression::Column { name, .. } => {
                let column_index = column_names.iter()
                    .position(|col| col == name)
                    .ok_or_else(|| Error::ColumnNotFound(name.clone()))?;
                
                Ok(row.get(column_index).cloned().unwrap_or(Value::Null))
            }
            Expression::Literal(literal) => {
                Ok(self.literal_to_value(literal))
            }
            Expression::BinaryOp { left, operator, right } => {
                let left_value = self.evaluate_expression(left, row, column_names)?;
                let right_value = self.evaluate_expression(right, row, column_names)?;
                self.evaluate_binary_op(&left_value, operator, &right_value)
            }
            Expression::Wildcard => {
                Err(Error::query_execution("通配符不能在此上下文中使用"))
            }
        }
    }
    
    /// 计算谓词表达式，返回布尔值
    fn evaluate_predicate(
        &self,
        expr: &Expression,
        row: &[Value],
        column_names: &[String],
    ) -> Result<bool> {
        let value = self.evaluate_expression(expr, row, column_names)?;
        match value {
            Value::Boolean(b) => Ok(b),
            _ => Err(Error::query_execution("谓词表达式必须返回布尔值")),
        }
    }

    /// 字面值转换为值
    fn literal_to_value(&self, literal: &LiteralValue) -> Value {
        match literal {
            LiteralValue::Integer(i) => Value::Integer(*i),
            LiteralValue::Float(f) => Value::Real(*f),
            LiteralValue::String(s) => Value::Text(s.clone()),
            LiteralValue::Boolean(b) => Value::Boolean(*b),
            LiteralValue::Null => Value::Null,
        }
    }
    
    /// 计算二元操作
    fn evaluate_binary_op(
        &self,
        left: &Value,
        operator: &BinaryOperator,
        right: &Value,
    ) -> Result<Value> {
        match (left, right) {
            (Value::Integer(l), Value::Integer(r)) => {
                match operator {
                    BinaryOperator::Plus => Ok(Value::Integer(l + r)),
                    BinaryOperator::Minus => Ok(Value::Integer(l - r)),
                    BinaryOperator::Multiply => Ok(Value::Integer(l * r)),
                    BinaryOperator::Divide => {
                        if *r == 0 {
                            Err(Error::query_execution("除零错误"))
                        } else {
                            Ok(Value::Integer(l / r))
                        }
                    }
                    BinaryOperator::Equal => Ok(Value::Boolean(l == r)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean(l != r)),
                    BinaryOperator::LessThan => Ok(Value::Boolean(l < r)),
                    BinaryOperator::LessThanOrEqual => Ok(Value::Boolean(l <= r)),
                    BinaryOperator::GreaterThan => Ok(Value::Boolean(l > r)),
                    BinaryOperator::GreaterThanOrEqual => Ok(Value::Boolean(l >= r)),
                    _ => Err(Error::query_execution("不支持的整数操作")),
                }
            }
            (Value::Boolean(l), Value::Boolean(r)) => {
                match operator {
                    BinaryOperator::And => Ok(Value::Boolean(*l && *r)),
                    BinaryOperator::Or => Ok(Value::Boolean(*l || *r)),
                    BinaryOperator::Equal => Ok(Value::Boolean(l == r)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean(l != r)),
                    _ => Err(Error::query_execution("不支持的布尔操作")),
                }
            }
            _ => Err(Error::query_execution("不支持的操作数类型")),
        }
    }
    
    /// 添加表数据（用于测试）
    pub fn add_table_data(&mut self, table_name: String, data: TableData) {
        self.table_data.insert(table_name, data);
    }
}

impl Default for QueryExecutor {
    fn default() -> Self {
        Self::new()
    }
}

/// 执行结果
#[derive(Debug, Clone)]
struct ExecutionResult {
    rows: Vec<Vec<Value>>,
    column_names: Vec<String>,
    affected_rows: usize,
}

/// 表数据
#[derive(Debug, Clone)]
pub struct TableData {
    pub rows: Vec<Vec<Value>>,
    pub column_names: Vec<String>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sql::planner::{Expression, LiteralValue};
    
    #[tokio::test]
    async fn test_executor_creation() {
        let executor = QueryExecutor::new();
        assert!(executor.table_data.is_empty());
        assert!(executor.vectorization_config.enabled);
        assert_eq!(executor.vectorization_config.batch_size, 1024);
    }

    #[test]
    fn test_vectorization_config() {
        let config = VectorizationConfig::default();
        assert!(config.enabled);
        assert_eq!(config.batch_size, 1024);
        assert_eq!(config.parallelism, 4);

        let custom_config = VectorizationConfig {
            batch_size: 512,
            enabled: false,
            parallelism: 2,
        };

        let executor = QueryExecutor::with_config(custom_config.clone());
        assert_eq!(executor.get_vectorization_config().batch_size, 512);
        assert!(!executor.get_vectorization_config().enabled);
    }

    #[tokio::test]
    async fn test_vectorized_table_scan() {
        let mut executor = QueryExecutor::new();

        // 添加测试数据
        let table_data = TableData {
            column_names: vec!["id".to_string(), "name".to_string()],
            rows: vec![
                vec![Value::Integer(1), Value::Text("Alice".to_string())],
                vec![Value::Integer(2), Value::Text("Bob".to_string())],
                vec![Value::Integer(3), Value::Text("Charlie".to_string())],
            ],
        };
        executor.table_data.insert("users".to_string(), table_data);

        // 测试向量化表扫描
        let result = executor.vectorized_table_scan("users", None, None).await.unwrap();
        assert_eq!(result.rows.len(), 3);
        assert_eq!(result.column_names, vec!["id", "name"]);

        // 测试带投影的向量化表扫描
        let projection = vec!["name".to_string()];
        let result = executor.vectorized_table_scan("users", None, Some(&projection)).await.unwrap();
        assert_eq!(result.rows.len(), 3);
        assert_eq!(result.column_names, vec!["name"]);
        assert_eq!(result.rows[0], vec![Value::Text("Alice".to_string())]);
    }

    #[tokio::test]
    async fn test_vectorized_filter() {
        let executor = QueryExecutor::new();

        let input = ExecutionResult {
            rows: vec![
                vec![Value::Integer(1), Value::Text("Alice".to_string())],
                vec![Value::Integer(2), Value::Text("Bob".to_string())],
                vec![Value::Integer(3), Value::Text("Charlie".to_string())],
            ],
            column_names: vec!["id".to_string(), "name".to_string()],
            affected_rows: 0,
        };

        // 创建一个简单的过滤条件 (id > 1)
        let predicate = Expression::BinaryOp {
            left: Box::new(Expression::Column {
                name: "id".to_string(),
                table: None,
            }),
            operator: BinaryOperator::GreaterThan,
            right: Box::new(Expression::Literal(LiteralValue::Integer(1))),
        };

        let result = executor.vectorized_filter(input, &predicate).await.unwrap();
        assert_eq!(result.rows.len(), 2); // 应该过滤掉id=1的行
    }

    #[tokio::test]
    async fn test_vectorized_limit() {
        let executor = QueryExecutor::new();

        let input = ExecutionResult {
            rows: vec![
                vec![Value::Integer(1)],
                vec![Value::Integer(2)],
                vec![Value::Integer(3)],
                vec![Value::Integer(4)],
                vec![Value::Integer(5)],
            ],
            column_names: vec!["id".to_string()],
            affected_rows: 0,
        };

        let result = executor.vectorized_limit(input, 3).await.unwrap();
        assert_eq!(result.rows.len(), 3);
    }

    #[tokio::test]
    async fn test_parallel_table_scan() {
        let mut executor = QueryExecutor::with_config(VectorizationConfig {
            batch_size: 1024,
            enabled: true,
            parallelism: 2,
        });

        // 添加测试数据
        let table_data = TableData {
            column_names: vec!["id".to_string(), "name".to_string()],
            rows: vec![
                vec![Value::Integer(1), Value::Text("Alice".to_string())],
                vec![Value::Integer(2), Value::Text("Bob".to_string())],
                vec![Value::Integer(3), Value::Text("Charlie".to_string())],
                vec![Value::Integer(4), Value::Text("David".to_string())],
            ],
        };
        executor.table_data.insert("users".to_string(), table_data);

        // 测试并行表扫描
        let result = executor.parallel_table_scan("users", None, None).await.unwrap();
        assert_eq!(result.rows.len(), 4);
        assert_eq!(result.column_names, vec!["id", "name"]);
    }

    #[tokio::test]
    async fn test_parallel_filter() {
        let executor = QueryExecutor::with_config(VectorizationConfig {
            batch_size: 1024,
            enabled: true,
            parallelism: 2,
        });

        let input = ExecutionResult {
            rows: vec![
                vec![Value::Integer(1), Value::Text("Alice".to_string())],
                vec![Value::Integer(2), Value::Text("Bob".to_string())],
                vec![Value::Integer(3), Value::Text("Charlie".to_string())],
                vec![Value::Integer(4), Value::Text("David".to_string())],
            ],
            column_names: vec!["id".to_string(), "name".to_string()],
            affected_rows: 0,
        };

        // 创建一个简单的过滤条件
        let predicate = Expression::BinaryOp {
            left: Box::new(Expression::Column {
                name: "id".to_string(),
                table: None,
            }),
            operator: BinaryOperator::GreaterThan,
            right: Box::new(Expression::Literal(LiteralValue::Integer(2))),
        };

        let result = executor.parallel_filter(input, &predicate).await.unwrap();
        // 注意：简化的并行过滤实现不会真正过滤，所以仍然是4行
        assert_eq!(result.rows.len(), 4);
    }

    #[tokio::test]
    async fn test_parallel_projection() {
        let executor = QueryExecutor::with_config(VectorizationConfig {
            batch_size: 1024,
            enabled: true,
            parallelism: 2,
        });

        let input = ExecutionResult {
            rows: vec![
                vec![Value::Integer(1), Value::Text("Alice".to_string())],
                vec![Value::Integer(2), Value::Text("Bob".to_string())],
            ],
            column_names: vec!["id".to_string(), "name".to_string()],
            affected_rows: 0,
        };

        let expressions = vec![
            Expression::Column {
                name: "name".to_string(),
                table: None,
            },
        ];

        let result = executor.parallel_projection(input, &expressions).await.unwrap();
        assert_eq!(result.rows.len(), 2);
        assert_eq!(result.column_names, vec!["name"]);
        assert_eq!(result.rows[0], vec![Value::Text("Alice".to_string())]);
    }

    #[test]
    fn test_parallel_configuration() {
        let config = VectorizationConfig {
            batch_size: 512,
            enabled: true,
            parallelism: 8,
        };

        let executor = QueryExecutor::with_config(config);
        assert_eq!(executor.get_vectorization_config().parallelism, 8);
        assert!(executor.get_vectorization_config().enabled);
    }
    
    #[test]
    fn test_literal_to_value() {
        let executor = QueryExecutor::new();
        
        assert_eq!(
            executor.literal_to_value(&LiteralValue::Integer(42)),
            Value::Integer(42)
        );
        
        assert_eq!(
            executor.literal_to_value(&LiteralValue::String("test".to_string())),
            Value::Text("test".to_string())
        );
        
        assert_eq!(
            executor.literal_to_value(&LiteralValue::Boolean(true)),
            Value::Boolean(true)
        );
    }
    
    #[test]
    fn test_binary_operations() {
        let executor = QueryExecutor::new();
        
        // 测试整数加法
        let result = executor.evaluate_binary_op(
            &Value::Integer(5),
            &BinaryOperator::Plus,
            &Value::Integer(3),
        ).unwrap();
        assert_eq!(result, Value::Integer(8));
        
        // 测试比较操作
        let result = executor.evaluate_binary_op(
            &Value::Integer(5),
            &BinaryOperator::GreaterThan,
            &Value::Integer(3),
        ).unwrap();
        assert_eq!(result, Value::Boolean(true));
        
        // 测试布尔操作
        let result = executor.evaluate_binary_op(
            &Value::Boolean(true),
            &BinaryOperator::And,
            &Value::Boolean(false),
        ).unwrap();
        assert_eq!(result, Value::Boolean(false));
    }
}
